package org.example.controller;

import org.example.service.AsyncService;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.listener.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@RestController
@RequestMapping("/sse")
public class SseController implements ApplicationContextAware {

    private static final Logger log = LoggerFactory.getLogger(SseController.class);
    private final ExecutorService executor = Executors.newCachedThreadPool();
    private final RTopic statusTopic;
    private final RedissonClient redissonClient;

    @Autowired
    public SseController(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
        this.statusTopic = redissonClient.getTopic("statusTopic");
    }

    @GetMapping(value = "/submit", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public String submitForReview() {
        executor.execute(() -> {
            try {
                statusTopic.publish("已提交");
                //Thread.sleep(2000);
                statusTopic.publish("排队中");
                //Thread.sleep(2000);
                statusTopic.publish("处理中");
               // Thread.sleep(2000);
                statusTopic.publish("已完成");
            } catch (Exception e) {
                log.error("任务处理出错", e);
            }
        });
        return "任务已提交";
    }

    @GetMapping(value = "/status", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
//       @CrossOrigin(origins = "http://localhost:5173")
    public SseEmitter streamStatus() {
        SseEmitter emitter = new SseEmitter(36000*1000L);
        log.info("SSE emitter created time" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        executor.execute(() -> {
            MessageListener<String> messageListener = (charSequence, message) -> {
                try {
                    message = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + message;
                    log.info("Received status update: {}", message);
                    sendStatus(emitter, message);
                } catch (Exception e) {
                    log.error("发送sse异常",e);
                    emitter.completeWithError(e);

                }
            };
            statusTopic.addListener(String.class, messageListener);
            emitter.onCompletion(() -> {
                log.info("SSE emitter completed, removing message listener");
                statusTopic.removeListener(messageListener);
            });
            emitter.onTimeout(() -> {
                log.info("SSE emitter timed out, removing message listener");
                log.info("Time Out Time" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                statusTopic.removeListener(messageListener);
            });
            emitter.onError((e) -> {
                log.info("SSE emitter encountered an error: {}, removing message listener", e.getMessage());
                statusTopic.removeListener(messageListener);
            });
        });
        return emitter;
    }

    private void sendStatus(SseEmitter emitter, String status) throws IOException {
        emitter.send(SseEmitter.event()
                .data(status)
                .name("statusUpdate").build());
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        MyController.asyncService= (AsyncService) applicationContext.getBean("asyncService");
    }
}
   